
package com.huaweicloud.haydn.delivercore.agent.service.metadata.impl;

import static com.huaweicloud.haydn.delivercore.agent.common.exception.BusinessErrorEnum.BASE_BAD_REQUEST;
import static com.huaweicloud.haydn.delivercore.agent.common.exception.BusinessErrorEnum.BASE_INTERNAL_SERVER_ERROR;

import com.alibaba.fastjson.JSON;
import com.huaweicloud.haydn.delivercore.agent.common.exception.BusinessException;
import com.huaweicloud.haydn.delivercore.agent.dto.DatalinkDatasetDto;
import com.huaweicloud.haydn.delivercore.agent.dto.excel.DbExcelDto;
import com.huaweicloud.haydn.delivercore.agent.dto.excel.FieldEmptyExcelDto;
import com.huaweicloud.haydn.delivercore.agent.dto.excel.FieldExcelDto;
import com.huaweicloud.haydn.delivercore.agent.dto.excel.TableExcelDto;
import com.huaweicloud.haydn.delivercore.agent.factory.ThreadPoolFactory;
import com.huaweicloud.haydn.delivercore.agent.service.CommonUtils;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.stream.Collectors;

/**
 * @since 2022-10-17
 */
@Slf4j
@Service("oracle")
public class OracleDatabaseUtil extends BaseDatabaseUtil {

    @Autowired
    private CommonUtils commonUtils;

    private static final ExecutorService EXECUTOR_SERVICE;

    private static final String FIELD_EMPTY_RATE_SQL =
        "select sum(case when length(?1) < 1 or ?2 is null then 0 else 1 end)/ count(*) empty_rate from ?3";

    static {
        EXECUTOR_SERVICE = ThreadPoolFactory.defaultExecutorService();
    }

    @Override
    public String getDriverClass() {
        return "oracle.jdbc.driver.OracleDriver";
    }

    @Override
    public String getSelectedDatabasesSql(String dbNames) throws BusinessException {
        return null;
    }

    @Override
    public String getAllDatabasesSql() {
        return "";
    }

    @Override
    public String getAllTablesSql() {
        return "select t.*, a.num_rows, a.avg_row_len, a.num_rows * a.avg_row_len from "
            + "(select t.tablespace_name,t.owner, t.segment_name, t.segment_type, sum(t.bytes/1024/1024) mb "
            + "from dba_segments t where t.segment_type='TABLE' "
            + "group by t.tablespace_name,t.OWNER, t.segment_name, t.segment_type) "
            + "t, dba_tables a where a.table_name = t.segment_name order by t.mb desc";
    }

    @Override
    public String getAllTablesByDbSql(String schemaNames) {
        return "select t.*, a.num_rows, a.avg_row_len, a.num_rows*a.avg_row_len from ("
            + "  select t.tablespace_name,t.owner, t.segment_name, t.segment_type, sum(t.bytes / 1024 / 1024) mb "
            + "  from dba_segments t where t.segment_type='TABLE' "
            + "  group by t.tablespace_name,t.OWNER, t.segment_name, t.segment_type"
            + ") t, dba_tables a where a.table_name = t.segment_name and a.owner in " + schemaNames
            + " order by t.mb desc";
    }

    @Override
    public String getSelectedTableSql(String schemaNames, String tableNames) {
        return "select t.*, a.num_rows, a.avg_row_len, a.num_rows*a.avg_row_len from ("
            + "  select t.tablespace_name,t.owner, t.segment_name, t.segment_type, sum(t.bytes / 1024 / 1024) mb "
            + "  from dba_segments t where t.segment_type='TABLE' "
            + "  group by t.tablespace_name,t.OWNER, t.segment_name, t.segment_type"
            + ") t, dba_tables a where a.table_name = t.segment_name and a.owner in " + schemaNames
            + " and t.segment_name in " + tableNames + " order by t.mb desc";
    }

    @Override
    public String getAllFieldsSql() {
        return null;
    }

    @Override
    public String getAllFieldsByDbSql(String dbNames) {
        return "";
    }

    @Override
    public String getAllFieldsSql(String schemaNames, String tableNames) {
        return "select c.owner, c.table_name as table_name, c.column_name, c.data_type, c.data_length, tempac.constraint_type, c.nullable, "
            + "c.data_precision, c.data_scale from dba_tab_columns c, dba_cons_columns consc, (select ac.owner as owner,ac.table_name as table_name, "
            + "ac.constraint_type as constraint_type, ac.constraint_name "
            + "as constraint_name from dba_constraints ac where   ac.constraint_type = 'P') tempac "
            + "where c.owner in " + schemaNames + "  and c.table_name in " + tableNames
            + " and c.owner= consc.owner (+)  and consc.owner = tempac.owner (+) "
            + "and c.table_name = consc.table_name (+) and consc.table_name= tempac.table_name (+) "
            + "and c.column_name  = consc.column_name (+) and consc.constraint_name = tempac.constraint_name (+)  "
            + "order by  table_name,column_name,constraint_type asc";
    }

    @Override
    public String getURL(String hostname, String port, String databaseName) {
        log.info("jdbc:oracle:thin:@" + hostname + ":" + port + ":" + databaseName);
        return "jdbc:oracle:thin:@" + hostname + ":" + port + ":" + databaseName;
    }

    @Override
    public String getURLException(String hostname, String port, String databaseName) {
        log.info("jdbc:oracle:thin:@" + hostname + ":" + port + "/" + databaseName);
        return "jdbc:oracle:thin:@" + hostname + ":" + port + "/" + databaseName;
    }

    private String getDbCapacity() {
        return "select sum(t.mb), count(*) from (select sum(t.bytes / 1024 / 1024) mb from dba_segments t "
            + "where t.segment_type='TABLE' group by t.tablespace_name,t.OWNER, t.segment_name, t.segment_type "
            + ") t";
    }

    private String genEmptyRate(String realFieldName, String realTableName) {
        return "select round(sum(case when length(" + realFieldName + ") < 1 or " + realFieldName
            + " is null then 1 else 0 end)/ count(*), 3) empty_rate from " + realTableName;
    }

    @Override
    public List<DbExcelDto> getAllDbs(String datalinkName, String dbNames, String schemas) {
        PreparedStatement ps = null;
        ResultSet rs = null;
        Connection conn = null;
        List<DbExcelDto> result = new ArrayList<>();
        try {
            SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
            String dbCapacitySql = getDbCapacity();
            log.info(dbCapacitySql);

            conn = getConnection();
            ps = conn.prepareStatement(dbCapacitySql, ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_READ_ONLY);
            rs = ps.executeQuery();
            while (rs.next()) {
                DbExcelDto dbExcelDto = new DbExcelDto();
                dbExcelDto.setDatabaseCapacity(rs.getString(1));
                dbExcelDto.setDatabaseTableNum(rs.getString(2));
                dbExcelDto.setDatabaseName(dbNames);
                dbExcelDto.setDatalinkName(datalinkName);
                String today = sdf.format(new Date());
                dbExcelDto.setGenTime(today);
                result.add(dbExcelDto);
            }
            return result;
        } catch (SQLException e) {
            log.error(e.getMessage());
            throw new BusinessException(BASE_INTERNAL_SERVER_ERROR);
        } finally {
            close(rs, ps, conn);
        }
    }

    @Override
    @SneakyThrows
    public List<TableExcelDto> getAllTables(String datalinkName, String dbNames, String schemas, String tableNames) {
        PreparedStatement ps = null;
        PreparedStatement psView = null;
        ResultSet rs = null;
        ResultSet rsView = null;
        Connection conn = null;

        List<TableExcelDto> result = new ArrayList<>();
        try {
            SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
            String today = sdf.format(new Date());
            String sql;
            String viewSql;
            String realSchemas = commonUtils.sqlQueryBuild(schemas);
            String realTableNames = commonUtils.sqlQueryBuild(tableNames);
            if (StringUtils.isNotBlank(realSchemas) && StringUtils.isNotBlank(realTableNames)) {
                sql = getSelectedTableSql(realSchemas, realTableNames);
                viewSql = getSelectedTableSqlView(realSchemas, realTableNames);
            } else if (StringUtils.isBlank(realSchemas) && StringUtils.isBlank(realTableNames)) {
                sql = getAllTablesSql();
                viewSql = getAllViewsSql();
            } else if (StringUtils.isNotBlank(realSchemas) && StringUtils.isBlank(realTableNames)) {
                sql = getAllTablesByDbSql(realSchemas);
                viewSql = getAllViewsByDbSql(realSchemas);
            } else {
                throw new BusinessException(BASE_INTERNAL_SERVER_ERROR,
                    "if you configured table.name, you must configure db.name as well");
            }

            log.info(sql);
            conn = getConnection();
            ps = conn.prepareStatement(sql, ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_READ_ONLY);
            rs = ps.executeQuery();
            while (rs.next()) {
                // 获取数据表中满足要求的一行数据，并放入Map中
                TableExcelDto tableExcelDto = new TableExcelDto();
                tableExcelDto.setGenTime(today);
                tableExcelDto.setDatalinkName(datalinkName);
                tableExcelDto.setDatabaseName(dbNames);
                tableExcelDto.setTableDescription("");
                tableExcelDto.setSchemaName(rs.getString(2));
                tableExcelDto.setTableName(rs.getString(3));
                tableExcelDto.setTablePhysicsCapacity(rs.getString(5));
                tableExcelDto.setTableRows(rs.getString(6));
                tableExcelDto.setTableAvgLength(rs.getString(7));
                tableExcelDto.setTableCapacity(rs.getString(8));
                result.add(tableExcelDto);
            }
            // Oracle中View和Table的信息分别存放在两张表里，所以需要再单独把View查询出来
            psView = conn.prepareStatement(viewSql, ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_READ_ONLY);
            rsView = psView.executeQuery();
            while (rsView.next()) {
                // 获取数据表中满足要求的一行数据，并放入Map中
                TableExcelDto tableExcelDto = new TableExcelDto();
                tableExcelDto.setGenTime(today);
                tableExcelDto.setDatalinkName(datalinkName);
                tableExcelDto.setDatabaseName(dbNames);
                tableExcelDto.setTableDescription("");
                tableExcelDto.setSchemaName(rsView.getString(1));
                tableExcelDto.setTableName(rsView.getString(2));
                result.add(tableExcelDto);
            }
            return result;
        } catch (SQLException e) {
            log.error(e.getMessage());
            throw new BusinessException(BASE_INTERNAL_SERVER_ERROR);
        } finally {
            close(rs, ps, conn);
            close(rsView, psView, null);
        }
    }

    private String getAllViewsByDbSql(String schemaNames) {
        return "SELECT dba_views.*  FROM dba_views  WHERE dba_views.owner IN " + schemaNames;
    }

    private String getAllViewsSql() {
        return "SELECT dba_views.*  FROM dba_views";
    }

    private String getSelectedTableSqlView(String schemaNames, String tableNames) {
        return "SELECT dba_views.*  FROM dba_views  WHERE dba_views.owner IN " + schemaNames
            + " and dba_views.view_name in " + tableNames;
    }

    @Override
    public List<FieldExcelDto> getAllFields(String datalinkName, String dbNames, String schemas, String tableNames) {
        log.info("service gen field data start now");
        String realSchemaNames = commonUtils.sqlQueryBuild(schemas);
        String realTableNames = commonUtils.sqlQueryBuild(tableNames);
        if (StringUtils.isBlank(realSchemaNames)) {
            throw new BusinessException(BASE_INTERNAL_SERVER_ERROR,
                "if the database type is oracle, you must fill schema.names");
        }
        Connection conn = null;
        try {
            List<TableExcelDto> allTables = getAllTables(datalinkName, dbNames, schemas, tableNames);
            List<String> tableNameList =
                allTables.stream().map(TableExcelDto::getTableName).collect(Collectors.toList());
            if (CollectionUtils.isEmpty(tableNameList)) {
                throw new BusinessException(BASE_INTERNAL_SERVER_ERROR, "Get empty tables");
            }
            conn = getConnection();
            if (StringUtils.isBlank(realTableNames)) {
                List<FieldExcelDto> allRes = new ArrayList<>();
                int divided = tableNameList.size() / 25;
                log.info("field loop start");
                CountDownLatch cdl = new CountDownLatch(divided + 1);
                log.info("divided size: " + divided);
                for (int i = 0; i <= divided; i++) {
                    List<String> subTableNames;
                    if (i == divided) {
                        subTableNames = tableNameList.subList(i * 25, tableNameList.size());
                    } else {
                        subTableNames = tableNameList.subList(i * 25, (i + 1) * 25);
                    }

                    String subTableStrings =
                        JSON.toJSONString(subTableNames).replace("\"", "'").replace("[", "(").replace("]", ")");
                    String sql = getAllFieldsSql(realSchemaNames, subTableStrings);
                    EXECUTOR_SERVICE.execute(execTask(allRes, dbNames, datalinkName, sql, cdl));
                }
                cdl.await();
                return allRes;
            } else {
                return getSubFields(conn, dbNames, datalinkName, getAllFieldsSql(realSchemaNames, realTableNames));
            }
        } catch (SQLException | InterruptedException e) {
            log.error(e.getMessage());
            return new ArrayList<>();
        } finally {
            close(null, null, conn);
        }
    }

    @Override
    public List<FieldEmptyExcelDto> getAllEmptyRateFields(String datalinkName, String dbNames, String schemas,
        Map<String, List<String>> fieldTableMap) {
        List<FieldEmptyExcelDto> resList = new ArrayList<>();
        if (fieldTableMap == null || fieldTableMap.size() == 0) {
            return resList;
        }
        List<String> tables = new ArrayList<>();
        fieldTableMap.forEach((k, v) -> tables.add(k));
        StringBuilder tableSb = new StringBuilder();
        for (int i = 0; i < tables.size(); i++) {
            String tableName = tables.get(i);
            tableSb.append(tableName);
            if (i != tables.size() - 1) {
                tableSb.append(",");
            }
        }
        String tableStr = tableSb.toString();
        List<TableExcelDto> allTables = getAllTables(datalinkName, dbNames, schemas, tableStr);
        Map<String, String> tableSchemaMap = new HashMap<>();
        allTables.forEach(dto -> tableSchemaMap.put(dto.getTableName(), dto.getSchemaName()));
        Connection conn = getConnection();

        for (String tableName : fieldTableMap.keySet()) {
            if (!tableSchemaMap.containsKey(tableName)) {
                continue;
            }
            List<String> fields = fieldTableMap.get(tableName);
            String schemaName = tableSchemaMap.get(tableName);
            String realTableName = schemaName + "." + tableName;
            fields.forEach(fieldName -> {
                String realFieldName = schemaName + "." + tableName + "." + fieldName;
                DatalinkDatasetDto datasetDto = new DatalinkDatasetDto();
                datasetDto.setDatalinkName(datalinkName);
                datasetDto.setDbName(dbNames);
                datasetDto.setSchemaName(schemaName);
                datasetDto.setTableName(tableName);
                datasetDto.setFieldName(fieldName);
                resList.addAll(getEmptyRateExcelDto(conn, realTableName, realFieldName, datasetDto));
            });
        }
        try {
            conn.close();
        } catch (SQLException e) {
            log.error(e.getMessage());
        }
        return resList;
    }

    private List<FieldEmptyExcelDto> getEmptyRateExcelDto(Connection conn, String realTableName, String realFieldName,
        DatalinkDatasetDto datasetDto) {
        PreparedStatement ps = null;
        ResultSet rs = null;
        List<FieldEmptyExcelDto> resList = new ArrayList<>();
        try {
            String emptyRateSql = genEmptyRate(realFieldName, realTableName);
            log.info(emptyRateSql);
            ps = conn.prepareStatement(emptyRateSql, ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_READ_ONLY);
            rs = ps.executeQuery();
            while (rs.next()) {
                FieldEmptyExcelDto fieldEmptyExcelDto = new FieldEmptyExcelDto();
                fieldEmptyExcelDto.setDatalinkName(datasetDto.getDatalinkName());
                fieldEmptyExcelDto.setDatabaseName(datasetDto.getDbName());
                fieldEmptyExcelDto.setSchemaName(datasetDto.getSchemaName());
                fieldEmptyExcelDto.setTableName(datasetDto.getTableName());
                fieldEmptyExcelDto.setFieldNameEn(datasetDto.getFieldName());
                String emptyRate = Optional.ofNullable(rs.getString(1)).orElse("1");
                log.info(emptyRate);
                fieldEmptyExcelDto.setEmptyRate(emptyRate);
                resList.add(fieldEmptyExcelDto);
            }
            return resList;
        } catch (SQLException e) {
            log.error(e.getMessage());
            return new ArrayList<>();
        } finally {
            close(rs, ps, null);
        }
    }

    private Runnable execTask(List<FieldExcelDto> allRes, String dbNames, String datalinkName, String sql,
        CountDownLatch cdl) {
        return () -> {
            Connection conn = null;
            try {
                conn = getConnection();
                allRes.addAll(getSubFields(conn, dbNames, datalinkName, sql));
                cdl.countDown();
                log.info("count down latch get count: " + cdl.getCount());
            } catch (SQLException e) {
                log.error(e.getMessage());
                throw new BusinessException(BASE_INTERNAL_SERVER_ERROR);
            } finally {
                close(null, null, conn);
            }
        };
    }

    private List<FieldExcelDto> getSubFields(Connection conn, String dbNames, String datalinkName, String sql)
        throws SQLException {
        PreparedStatement ps = null;
        ResultSet rs = null;
        try {
            SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
            String today = sdf.format(new Date());
            ps = conn.prepareStatement(sql, ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_READ_ONLY);
            rs = ps.executeQuery();
            log.info(sql);
            List<FieldExcelDto> result = new ArrayList<>();
            while (rs.next()) {
                FieldExcelDto fieldExcelDto = new FieldExcelDto();
                fieldExcelDto.setGenTime(today);
                fieldExcelDto.setDatalinkName(datalinkName);
                fieldExcelDto.setDatabaseName(dbNames);
                fieldExcelDto.setSchemaName(rs.getString(1));
                fieldExcelDto.setTableName(rs.getString(2));
                fieldExcelDto.setFieldNameEn(rs.getString(3));

                fieldExcelDto.setFieldType(rs.getString(4));
                fieldExcelDto.setFieldLength(rs.getString(5));

                if ("P".equals(rs.getString(6))) {
                    fieldExcelDto.setIsPrimaryKey("YES");
                }
                if ("Y".equals(rs.getString(7))) {
                    fieldExcelDto.setIsNullable("YES");
                }
                fieldExcelDto.setDataPrecision(rs.getString(8));
                fieldExcelDto.setDataScale(rs.getString(9));
                result.add(fieldExcelDto);
            }
            if (CollectionUtils.isNotEmpty(result)) {
                log.info("result[0]: " + result.get(0));
            } else {
                log.info("empty result at getSubFields");
            }

            return result;
        } catch (SQLException sqlException) {
            log.error(sqlException.getMessage());
            throw new BusinessException(BASE_BAD_REQUEST);
        } finally {
            if (ps != null) {
                ps.close();
            }
            if (rs != null) {
                rs.close();
            }
        }
    }
}
